package com.econ.powercloud.ems.timing.report.prometheus.callback;

import com.econ.powercloud.console.facade.utils.CacheKeyUtils;
import com.econ.powercloud.ems.common.util.JacksonExtUtil;
import com.econ.powercloud.ems.util.AsyncHttpClientUtil;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * kafka每分钟消息数
 */
@Component
public class PrometheusKafkaTopicMinConsumeCallback extends AbstractPrometheusMultiDataCallback implements AsyncHttpClientUtil.StringFutureCallback {

    final private String category = "kafkaTopic";

    final private String ITEM_TEMPLATE = "minConsume#%s";

    final private String TOPIC_KEY_TEMPLATE = "data.result[%s].metric.topic";

    @Override
    public void success(String content) {
        this.dealWithPrometheusCallbackData(content);
    }

    @Override
    public void dealWithPrometheusCallbackData(String content) {
        Map<String, String> prometheusResult = JacksonExtUtil.flatJsonToMap(content);
        for(int i = 0; i < super.MAX_INSTANCE_COUNT ; i++){
            String instanceKey = getInstanceKey(i);
            String topicKey = String.format(TOPIC_KEY_TEMPLATE, i);
            if(!prometheusResult.containsKey(instanceKey)){
                break;
            }
            String topic = prometheusResult.get(topicKey);
            String instance = (String)prometheusResult.get(instanceKey).replaceAll(":", "-");
            String value = prometheusResult.get(getValueKey(i));
            String redisKey = CacheKeyUtils.getMonitorDataKey(instance, getCategory());
            HashOperations<String, String, String> hashOperations = super.stringRedisTemplate.opsForHash();
            hashOperations.put(redisKey, getItem(topic), value);
        }
    }

    @Override
    String getCategory() {
        return this.category;
    }

    @Override
    String getItem() {
        return null;
    }

    String getItem(String topic) {
        return String.format(ITEM_TEMPLATE, topic);
    }
}
